package demo;

import org.apache.flink.api.java.tuple.Tuple2;
import org.apache.flink.streaming.api.datastream.DataStream;
import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;
import org.apache.flink.table.api.Table;
import org.apache.flink.table.api.TableResult;
import org.apache.flink.table.api.bridge.java.StreamTableEnvironment;
import org.apache.flink.types.Row;

/**
 * @author ChinaManor
 * #Description T3
 * #Date: 24/6/2021 13:06
 */
public class T3 {
    public static void main(String[] args) throws Exception {
        //1.准备环境 流执行环境和流表
        StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();
        StreamTableEnvironment tEnv = StreamTableEnvironment.create(env);

//2.执行SQL,创建 input_kafka 表
        TableResult inputTable = tEnv.executeSql(
                "CREATE TABLE CategoryPojo (\n" +
                        "  `category` STRING,\n" +
                        "  `totalPrice` BIGINT,\n" +
                        "  `dateTime` STRING\n" +
                        ") WITH (\n" +
                        "  'connector' = 'kafka',\n" +
                        "  'topic' = 'CategoryPojo',\n" +
                        "  'properties.bootstrap.servers' = 'node1:9092',\n" +
                        "  'properties.group.id' = 'testGroup',\n" +
                        "  'scan.startup.mode' = 'latest-offset',\n" +
                        "  'format' = 'json'\n" +
                        ")"
        );
// 创建 output_kafka
        TableResult outputTable = tEnv.executeSql(
                "CREATE TABLE output_kafka (\n" +
                        "  `user_id` BIGINT,\n" +
                        "  `page_id` BIGINT,\n" +
                        "  `status` STRING\n" +
                        ") WITH (\n" +
                        "  'connector' = 'kafka',\n" +
                        "  'topic' = 'output_kafka',\n" +
                        "  'properties.bootstrap.servers' = 'node1:9092',\n" +
                        "  'format' = 'json',\n" +
                        "  'sink.partitioner' = 'round-robin'\n" +
                        ")"
        );

// 根据 status 是否为 success 条件筛选出来值
        String sql = "select " +
                "category," +
                "totalPrice,"+
                "dateTime  FROM CategoryPojo ";

        Table ResultTable = tEnv.sqlQuery(sql);
        //3.toRetractStream
        DataStream<Tuple2<Boolean, Row>> resultDS = tEnv.toRetractStream(ResultTable, Row.class);
        //4.打印输出
        resultDS.print();
        //5.执行sql 将筛选出来success的数据表插入到 output_kafka
//        tEnv.executeSql("insert into output_kafka select * from "+ResultTable);


        //6.excute
        env.execute();
    }
}
